﻿
#include "stdafx.h"
#include "iocp.h"
#include "netExtension.h"
#include "connector.h"
#include "networkModule.h"
#include "IoPacketPool.h"
#include "ssession.h"




//////////////////////////////////////////////////////////////////////////
// IocpRecv	
//////////////////////////////////////////////////////////////////////////
IocpRecv::IocpRecv()
{
	CmASSERT(nullptr!=NetworkModule::ms_pIoPacketPool);
	m_pIocpPacket = nullptr;
	m_pIocpPacketPrev= nullptr;
}

IocpRecv::~IocpRecv()
{
	CmASSERT(nullptr!=NetworkModule::ms_pIoPacketPool);
	//µî·ÏµÈ ¸ðµç ÆÐÅ¶ Á¦°Å
	if(m_pIocpPacket)		
		NetworkModule::ms_pIoPacketPool->ReleaseBuffer(m_pIocpPacket);

	if(m_pIocpPacketPrev)
		NetworkModule::ms_pIoPacketPool->ReleaseBuffer(m_pIocpPacketPrev);

	m_pIocpPacket = nullptr;
	m_pIocpPacketPrev = nullptr;
}

BYTE* IocpRecv::GetBuffer(int recvSize, int& fullsize)
{
	BYTE* pPacket=nullptr;
	if(m_pIocpPacketPrev)
	{
		IocpPacket* pNewBuffer = NetworkModule::ms_pIoPacketPool->GetRecvBuffer();			
		if( (m_pIocpPacketPrev->GetRawSize()+recvSize) > IoPacketPool::EACH_BUF_SIZE )
		{	//ÇÑ ¹öÆÛ°¡ ³Ê¹« Ä¿¼­ »õ·Î »ý¼º ¹Þ´Â´Ù. 
			delete [] pNewBuffer->m_pPacket;
			pNewBuffer->m_pPacket = new BYTE[m_pIocpPacketPrev->GetRawSize()+recvSize];
		}

		pNewBuffer->m_size = m_pIocpPacketPrev->GetRawSize()+recvSize;				

		BYTE* pPrevBuf=pNewBuffer->GetRawBuffer();
		memcpy(pPrevBuf, m_pIocpPacketPrev->GetRawBuffer()+m_pIocpPacketPrev->m_startPos, m_pIocpPacketPrev->GetRawSize() );
		pPrevBuf +=  m_pIocpPacketPrev->GetRawSize();
		memcpy(pPrevBuf, m_pIocpPacket->GetRawBuffer(), recvSize );

		pPacket = pNewBuffer->GetRawBuffer();
		fullsize = pNewBuffer->GetRawSize();

		NetworkModule::ms_pIoPacketPool->ReleaseBuffer(m_pIocpPacketPrev);
		m_pIocpPacketPrev = pNewBuffer;
	}
	else
	{
		m_pIocpPacket->m_size = recvSize;

		pPacket = m_pIocpPacket->GetRawBuffer();
		fullsize = recvSize;
	}

	return pPacket;
}


BYTE* IocpRecv::GetBufferPrev()
{
	if(m_pIocpPacketPrev)
		return m_pIocpPacketPrev->GetRawBuffer();
	return nullptr;
}

void IocpRecv::SetBuffer(WSABUF*	pWsaBuf)
{		
	//new¸¦ »ý¼ºÇØ¼­ postÇÑ´Ù. 
	ZeroMemory(pWsaBuf, sizeof(WSABUF));
	if(m_pIocpPacket)
	{
		NetworkModule::ms_pIoPacketPool->ReleaseBuffer(m_pIocpPacket);
	}

	m_pIocpPacket=NetworkModule::ms_pIoPacketPool->GetRecvBuffer();
	pWsaBuf->buf = (CHAR*)m_pIocpPacket->GetRawBuffer();
	pWsaBuf->len = m_pIocpPacket->GetRawSize();		
}

bool IocpRecv::CompletionRecv(int remain)
{
	if(m_pIocpPacket)
	{
		NetworkModule::ms_pIoPacketPool->ReleaseBuffer( m_pIocpPacket );
		m_pIocpPacket = nullptr;
	}

	if(0==remain)
	{
		if(m_pIocpPacketPrev)
		{
			NetworkModule::ms_pIoPacketPool->ReleaseBuffer( m_pIocpPacketPrev );
			m_pIocpPacketPrev = nullptr;
		}

		
		return true;
	}

	if(!m_pIocpPacketPrev)
	{
		m_pIocpPacketPrev=m_pIocpPacket;
	}
	else
	{
		NetworkModule::ms_pIoPacketPool->ReleaseBuffer( m_pIocpPacket );
	}

	m_pIocpPacketPrev->m_startPos = m_pIocpPacketPrev->m_size - remain;
	m_pIocpPacketPrev->m_size = remain;


	return true;
}

//////////////////////////////////////////////////////////////////////////
// IocpSend
//////////////////////////////////////////////////////////////////////////
IocpSend::IocpSend(SSession* pSession)
:m_pSession(pSession), m_pFirstIocpPacket(nullptr), m_pNextIocpPacket(nullptr), m_bComplete(false)
	,m_pReleseIocpPacket(nullptr)
	,m_pNextRelesePacket(nullptr)
	,m_sendLock()
{
	CmASSERT(nullptr!=NetworkModule::ms_pIoPacketPool);
}

IocpSend::~IocpSend()
{
	CmASSERT(nullptr!=NetworkModule::ms_pIoPacketPool);
}

void IocpSend::Send(unsigned long type, BYTE* pPacket, const size_t& size)
{
	IocpPacket* pIocpPacket = NetworkModule::ms_pIoPacketPool->Alloc();			
	CmASSERT(NULL!=pIocpPacket);

	pIocpPacket->SetPacket(type, pPacket, size);

	Lock();
	{
		if( m_pFirstIocpPacket!=NULL )
		{
			IocpPacket* pIocpPacketLast = m_pFirstIocpPacket;
			while(pIocpPacketLast->m_pNext)
			{
				pIocpPacketLast=pIocpPacketLast->m_pNext;
			}

			pIocpPacketLast->m_pNext = pIocpPacket;
		}
		else
		{
			m_pFirstIocpPacket = pIocpPacket;
		}
	}
	Unlock();

	//iocp¿¡¼­ º¸³»µµ·Ï ÇÑ´Ù. 
	//PushSendThread();
}

void IocpSend::PostSendDirect()
{
	IocpPacket* pIocpPacket =NULL;
	IocpPacket* pNextIocpPacket =NULL;

	Lock();
	{
		if( NULL!=m_pFirstIocpPacket )
		{
			m_pNextIocpPacket = m_pFirstIocpPacket;
			m_pFirstIocpPacket = NULL;
		}
	}		
	Unlock();

	pNextIocpPacket=m_pNextIocpPacket;		
	m_pNextIocpPacket=NULL;

	while(pNextIocpPacket)
	{
		pIocpPacket = pNextIocpPacket;
		PostSendEach( pIocpPacket->m_pPacket, pIocpPacket->m_size );
		AddRelesePacketList(pIocpPacket);

		pNextIocpPacket = pNextIocpPacket->m_pNext;		

	}
}

bool IocpSend::PostSendEach(BYTE* pPacket, const size_t& size)
{
	SetWaitSendComplete( true, size );
	WSABUF ws;
	ws.buf = (CHAR*)pPacket;
	ws.len = size;

	SSession::ST_IOCP_DATA* pIocpData = m_pSession->GetSendIocpData();
	CmASSERT(NULL!=pIocpData);       

	INT32 nNetErr = ::WSASend( m_pSession->GetSocket(), &ws, 1, &pIocpData->_size, pIocpData->_flags, (OVERLAPPED*)pIocpData, NULL );
	if( SOCKET_ERROR==nNetErr)
	{
		INT32 err= ::WSAGetLastError();
		if( WSA_IO_PENDING == err )
			return true;
		if( WSAEWOULDBLOCK == err )
			return true;
		//failed
		{
			//LOG_MGR__NORMAL( TEXT("ERR: OnPostSend %d %s \n"), 
			//	err, 
			//	NetUtil::GetLastErrorMessage( _SC<DWORD>(err) )                    
			//	);

			//if( 10055==err||10053==err)
			//{
			//	// ÀÌÀü »çÀÌÁî°¡ ÇöÀç »çÀÌÁî¸¦ ÃÊ°úÇØ¼­ º¸³»´ÂÁö È®ÀÎ
			//	LOG_MGR__NORMAL( TEXT("\t BufSize %d\n"), 
			//		Net::NetUtil::GetOptBufSize( sock, SO_SNDBUF) );
			//}                
			return false;
		}
	}	
	return true;
}


void IocpSend::OnSend(size_t size)
{
	Lock();

	if(nullptr!=m_pFirstIocpPacket)	//º¸³¾°ÍÀÌ ÀÖÀ¸¸é  postSend¸¦ °è¼ÓÇÑ´Ù. 
	{
		PostSendDirect();
	}
	else
	{
		SetWaitSendComplete( false, size );
	}

	Unlock();
}

void IocpSend::AddRelesePacketList(IocpPacket* pIocpPacket)
{
	Lock();
	{
		if( m_pReleseIocpPacket!=NULL )
		{
			IocpPacket* pIocpPacketLast = m_pReleseIocpPacket;
			while(pIocpPacketLast->m_pNext)
			{
				pIocpPacketLast=pIocpPacketLast->m_pNext;
			}

			pIocpPacketLast->m_pNext = pIocpPacket;
		}
		else
		{
			m_pReleseIocpPacket = pIocpPacket;
		}
	}
	Unlock();
}

void IocpSend::ReleseSendPacket()
{

	IocpPacket* pIocpPacket =NULL;
	IocpPacket* pNextIocpPacket =NULL;

	Lock();
	{
		if( NULL!=m_pReleseIocpPacket )
		{
			m_pNextRelesePacket = m_pReleseIocpPacket;
			m_pReleseIocpPacket = NULL;
		}
	}		
	Unlock();



	pNextIocpPacket=m_pNextRelesePacket;		
	m_pNextRelesePacket=NULL;

	while(pNextIocpPacket)
	{
		pIocpPacket = pNextIocpPacket;
		pIocpPacket->DeletePacket();
		pNextIocpPacket = pNextIocpPacket->m_pNext;
		NetworkModule::ms_pIoPacketPool->Free(pIocpPacket);
	}

}
void IocpSend::SetWaitSendComplete(const bool& bComplete, const int& postedSize)
{
	::InterlockedExchange( (long*)&m_bComplete, bComplete);

	// °É¾úÀ»¶§ ½Ã°£°ú Ç®¾úÀ»¶§ÀÇ ½Ã°£À» °è»êÇÑ´Ù 
	if( true==bComplete)
	{
		m_waittingCnt = 1;
		m_postSendSize += postedSize;
	}
	else
	{
		if(0>postedSize)
		{
			m_postSendSize= 0;
		}
		else
		{
			m_postSendSize -= postedSize;
			ReleseSendPacket();
		}
	}
}

bool IocpSend::IsWaitComplete()
{
	if( m_bComplete)
	{
		++m_waittingCnt;
	}

	return m_bComplete; 	
}

int	IocpSend::GetWaittingCnt() const
{		
	return m_waittingCnt;
}

int IocpSend::GetPostSendSize() const
{
	return m_postSendSize;
}

void IocpSend::Lock()
{
	m_sendLock.Lock();
}

void IocpSend::Unlock()
{
	m_sendLock.Unlock();
}

//////////////////////////////////////////////////////////////////////////
// IocpAccept
//////////////////////////////////////////////////////////////////////////
bool IocpAccept::Accept(const TCHAR* pIP, const int& port)
{	
	m_hAcceptIOCP = ::CreateIoCompletionPort( INVALID_HANDLE_VALUE, NULL, 0, 0 );

	if(!ListenSock(pIP, port))
	{
		CmTRACE( L"failed ListenSock \n");
		return false;
	}

	Extension::LoadExtensionFunction(m_acceptSocket);
	::CreateIoCompletionPort( (HANDLE)m_acceptSocket, m_hAcceptIOCP, (ULONG_PTR)&m_acceptSocket, 0 );

	DWORD dwAcceptSize=10;
	m_CurrentAcceptSize=0;
	m_MaxAcceptSize= dwAcceptSize+1;
	PostAcceptIOCP( m_MaxAcceptSize );

	std::wstring acceptName;
	{//Set Thread name
		std::wstringstream os;
		os << L"Accept " << port;
		acceptName = os.str();
	}

	HANDLE handle=NetworkModule::CreateNetThread(gRunAcceptThread, (LPVOID)this, acceptName );
	m_hAcceptThread= handle;		
	return true;
}

bool IocpAccept::ListenSock(const TCHAR* pIP, const int& port)
{
	UNREFERENCED_PARAMETER(pIP);

	m_acceptSocket = ::WSASocket( AF_INET, SOCK_STREAM, IPPROTO_IP, NULL, 0, WSA_FLAG_OVERLAPPED ); 
	
	ZeroMemory( &m_acceptAddr, sizeof( SOCKADDR_IN ) );
	m_acceptAddr.sin_family = AF_INET;
	if(!pIP) 
	{	// accept any
		m_acceptAddr.sin_addr.s_addr = htonl(INADDR_ANY);
	}
	else
	{
		//todo accept할때 IP지정
	}
	
	m_acceptAddr.sin_port = (int)::htons( (u_short)port);

	//disable TIME_WAIT
	//int reuse=1;
	//NetUtil::SetSockOpt( m_socket, SO_REUSEADDR, (char*)&reuse, sizeof(reuse) );

	if( SOCKET_ERROR==::bind( m_acceptSocket, (SOCKADDR*)&m_acceptAddr, sizeof( SOCKADDR_IN ) ) )
	{
		CmTRACE( TEXT("Fail: bind(%d) \n"), WSAGetLastError() );
		return false;
	}


	if( SOCKET_ERROR ==::listen( m_acceptSocket, NET_MAX_LISTENCON) )
	{
		CmTRACE( TEXT("Fail: listen(%d) \n"), WSAGetLastError() );
		return false;
	}

	return true;
}

void IocpAccept::PostAcceptIOCP(const DWORD& dwAccpetSize)
{		
	DWORD poolSize=0;
	SSession* pNewSession= NULL;        
	while( dwAccpetSize>poolSize )
	{		
		//pNewSession = new SSession;
		pNewSession = m_pParentSession->m_pNetworkModule->GetNewSession();
		PostAccpet( pNewSession );
		++poolSize;
	}		
}

bool IocpAccept::PostAccpet(SSession *pSession)
{
	DWORD dwBytes = 0;
	BYTE* pRecvBuf= NULL;
	
	pSession->m_pAcceptSession = m_pParentSession;

	SOCKET	pSocket = ::WSASocket( AF_INET, SOCK_STREAM, IPPROTO_IP, NULL, 0, WSA_FLAG_OVERLAPPED ); 
	pSession->SetSocket(pSocket);
	CmASSERT(NULL!=pSocket);
		
	SSession::ST_IOCP_ACCEPT *pAccpet = pSession->GetAccpIocpData();
	pRecvBuf=pAccpet->_pAcceptBuf;
	dwBytes=ACCEPTED_ADDR_SIZE_;


	if( SOCKET_ERROR==Extension::m_lpfnAccepteEx(
		m_acceptSocket, 
		pSocket , 
		pRecvBuf, 
		0,
		sizeof(SOCKADDR_IN) + 16,
		sizeof(SOCKADDR_IN) + 16,
		&dwBytes,
		pAccpet) )
	{
		int err = ::WSAGetLastError();            
		if( ERROR_IO_PENDING != err )
		{
			CmTRACE( TEXT("ERROR: PostAccpet [err==%d]\n"), err);
			//	NetUtil::GetLastErrorMessage( _SC<DWORD>(err) ) );
			return FALSE;
		}
	}

	//pSession->Create();// m_eConnectType, eSESSION_POOL_ACCEPT, E_SESSION_STATE_WAIT_ACCEPTTING);


	IncrementAcceptSize();

	return TRUE;
}

void IocpAccept::DestoryAccept()
{
	if(m_acceptSocket!=INVALID_SOCKET)
	{
		::shutdown(m_acceptSocket, SD_BOTH );
		::closesocket(m_acceptSocket);
		m_acceptSocket= INVALID_SOCKET;

		::PostQueuedCompletionStatus( m_hAcceptIOCP, 0, 0, NULL );        

		// ¸ðµç ¾²·¹µå°¡ Á¾·áÇÏ±â¸¦ ±â´Ù¸°´Ù. 
		//CmTime::Sleep(100);     // °­Á¦Á¾·áÇÏÁö ¾ÊÅä·Ï Á¶±Ý ´ë±âÇÑ´Ù.

		::WaitForSingleObject( m_hAcceptThread, INFINITE );
		::CloseHandle( m_hAcceptThread );
	}
}

int IocpAccept::ThreadAccept()
{
	BOOL bSuccess=false;
	SSession ::ST_IOCP_ACCEPT* pIocpAccept = nullptr;
	void *pAccetDumy = nullptr;
	DWORD dwIoSize = 0;
	SOCKADDR* pLocalAddr = nullptr;
	SOCKADDR* pRemoteAddr	= nullptr;
	int iLocalAddrLen = 0, iRemoteAddrLen = 0;


	bSuccess = ::GetQueuedCompletionStatus(m_hAcceptIOCP,
		&dwIoSize,
		reinterpret_cast<PULONG_PTR>(&pAccetDumy),
		reinterpret_cast<LPOVERLAPPED*>(&pIocpAccept),
		INFINITE);

	if( !bSuccess )
	{
		if( pAccetDumy != NULL && WSAGetLastError() == 64 ) 
		{
			CmTRACE( TEXT("INFO: pAccetDumy != NULL. \n") );
		}
		return -1;
	}

	if(NULL==pIocpAccept)
	{
		CmTRACE( TEXT("[ACCEPT THREAD] overLapped==NULL \n") );
		return -1;
	}

	if(NULL==pAccetDumy)
	{                
		CmTRACE( TEXT("[ACCEPT THREAD] iocpKey==NULL  \n") );
		return -1;            
	}

	switch( pIocpAccept->_type)
	{
	case SSession::eTYPE_ACCEPT:
		{
			SSession* pSession = pIocpAccept->_pSession;

			DecrementAcceptSize();
			if( 0>=GetCurrentAcceptSize())
			{
				CmTRACE( TEXT("[ ACCEPT ] SERVER IS Pool!! \n") );
				SOCKET sock=pSession->GetSocket();
				::shutdown( sock, SD_BOTH );
				::closesocket( sock );

				PostAccpet( pSession );
				return 1;
			}

			BYTE*	pBuf	= pIocpAccept->_pAcceptBuf;
			//DWORD	dwLength = ACCEPTED_ADDR_SIZE_;

			Extension::m_lpfnGetAcceptExSockAddrs( 
				pBuf, 
				0,
				sizeof( SOCKADDR_IN ) + 16,
				sizeof( SOCKADDR_IN ) + 16,
				&pLocalAddr,
				&iLocalAddrLen,
				&pRemoteAddr,
				&iRemoteAddrLen
				);

			//¹Ì¸® postÇÑ SessionÀÇ ¼¼ÆÃÀ» ´Ù½ÃÇÑ´Ù.                    
			pSession->Create();

			CmASSERT(pIocpAccept->_socket!=INVALID_SOCKET );
			CmVERIFY(pIocpAccept->_socket!=INVALID_SOCKET);


			pSession->OnAccept();

			//Accept¼º°øÈÄ Ã³¸®
			pSession->SetState(SSession::eREADY);
			pSession->AssociateIOCPKey();
			pSession->PostRecv();

			//post new session
			SSession* pNewSession = m_pParentSession->m_pNetworkModule->GetNewSession();
			PostAccpet( pNewSession );			
		}break;

	default:
		{
			CmASSERTFAIL_MSG( L"error" ) ;
		}break;
	}

	CmTRACE( TEXT("[ ACCEPT ] Shotdown MainIocpAccept. \n") );
	return 1;
}
